//package com.zyzg.listenter;
//
//import com.alibaba.fastjson2.JSONObject;
//import com.rabbitmq.client.Channel;
//import com.zyzg.constant.BusinessConstants;
//import com.zyzg.factory.ExecutorFactory;
//import com.zyzg.service.IDataService;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.rabbit.annotation.*;
//import org.springframework.amqp.support.AmqpHeaders;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.messaging.handler.annotation.Header;
//import org.springframework.scheduling.annotation.Async;
//import org.springframework.stereotype.Component;
//
//import java.nio.charset.StandardCharsets;
//import java.util.concurrent.ConcurrentLinkedQueue;
//import java.util.concurrent.ExecutorService;
//
///**
// * 数据监听类
// * @author rui
// */
//@Slf4j
//@Component
//public class BusinessMsgListener_bf {
//
////    private static final int MAX_TASKS_PER_SECOND = 700; // 每秒最大任务数
////    private static long lastSubmitTime = System.currentTimeMillis();
////    private static int taskCount = 0;
//
//    private static final ConcurrentLinkedQueue<JSONObject> queue_0200 = new ConcurrentLinkedQueue<>();
//    private static final ConcurrentLinkedQueue<JSONObject> queue_0704 = new ConcurrentLinkedQueue<>();
//
//    @Autowired
//    private IDataService dataService;
//
//
//    /**
//     * 设备位置数据消费
//     * @param message
//     * @return
//     */
//    @RabbitHandler
//    @RabbitListener(bindings = {@QueueBinding(
//            value = @Queue(value = BusinessConstants.JT808_LOCATION_QUEUE, durable = "true", declare = "true"),
//            exchange = @Exchange(value = BusinessConstants.JT808_LOCATION_EXCHANGE, type = "fanout", durable = "true", declare = "true"),
//            declare = "true")}, containerFactory = "businessContainerFactory")
//    public void processLocation(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
//        try {
//            String msgStr = new String(message.getBody(), StandardCharsets.UTF_8);
//            // log.info("BusinessMsgListener processLocation() 监听到一条消息:{}", msgStr);
//            JSONObject msg = JSONObject.parseObject(msgStr);
//            // 剩下业务逻辑处理
//            // 判断type字段，进行分类处理
//            String msgType = msg.getString("type");
//
//            processMessageAsync(msg, msgType, channel, deliveryTag);
//
//        } catch (Exception e) {
//            try {
//                // 消息处理失败，拒绝消息并重新入队
//                channel.basicNack(deliveryTag, false, true);
//            } catch (Exception nackException) {
//                nackException.printStackTrace();
//            }
//            e.printStackTrace();
//        }
//    }
//
//    @Async("messageProcessingExecutor")
//    public void processMessageAsync(JSONObject msg, String msgType, Channel channel, long deliveryTag) {
//        try {
//            ExecutorService executor = ExecutorFactory.getInstance();
//            if (msgType.equals("0x0200")) {
////                // 执行处理实时GPS数据逻辑， 分别处理实时数据、报警、历史数据、位置解析； 判断经纬度是否合法，不合法舍弃，判断是否为漂移点，漂移舍弃
////                submitTask(executor, () -> {
////                    // 处理实时GPS数据(缓存到redis，达到100条数据集中批量处理)
////                    dataService.handleRealMsg(msg);
////                });
////                submitTask(executor, () -> {
////                    // 处理报警数据(缓存到redis，达到100条数据集中批量处理)
////                    dataService.handleAlarmMsg(msg);
////                });
////                submitTask(executor, () -> {
////                    // 处理历史数据&位置解析(缓存到redis，达到1000条数据集中批量处理)
////                    dataService.handleHistoryAndAreaMsg(msg);
////                });
//                queue_0200.offer(msg);
//            } else if (msgType.equals("0x0704")) {
////                submitTask(executor, () -> {
////                    // 处理补报的实时GPS数据(缓存到redis，达到100条数据集中批量处理)
////                    dataService.handleRepairRealMsg(msg);
////                });
////                submitTask(executor, () -> {
////                    // 处理补报的报警数据(缓存到redis，达到100条数据集中批量处理)
////                    dataService.handleRepairAlarmMsg(msg);
////                });
////                submitTask(executor, () -> {
////                    // 处理补报的历史数据&位置解析(缓存到redis，达到1000条数据集中批量处理)
////                    dataService.handleRepairHistoryAndAreaMsg(msg);
////                });
//                queue_0704.offer(msg);
//            }
//
//            // 消息处理成功，手动确认消息
//            channel.basicAck(deliveryTag, false);
//        } catch (Exception e) {
//            try {
//                channel.basicNack(deliveryTag, false, true);
//            } catch (Exception nackException) {
//                nackException.printStackTrace();
//            }
//            e.printStackTrace();
//        }
//    }
//
////    public static void submitTask(ExecutorService executor, Runnable task) {
////        long currentTime = System.currentTimeMillis();
////        if (currentTime - lastSubmitTime >= 1000) {
////            lastSubmitTime = currentTime;
////            taskCount = 0;
////        }
////
////        if (taskCount < MAX_TASKS_PER_SECOND) {
////            executor.submit(task);
////            taskCount++;
////        } else {
////            // 任务提交速度过快，进行限流处理
////            try {
////                TimeUnit.MILLISECONDS.sleep(500); // 等待一段时间后再尝试提交
////                submitTask(executor, task);
////            } catch (InterruptedException e) {
////                e.printStackTrace();
////            }
////        }
////    }
//
//}
